import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StructField, StructType}
import org.junit.Test

class 缺失值的处理 {
  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("Test")
    .getOrCreate()

  @Test
  def null_and_Nan(): Unit = {
    //读取数据集
    /** 1、
      * 下面的存在一些问题，当推断数字类型的时候，遇到Nan会推断为字符串类型
      *spark.read
      * .option("header",true)
      * //类型推断
      * .option("inferredScheam",true)
      * .csv("C:\\Users\\HR\\Desktop\\DataAnalysis-master\\DataAnalysis-master\\day06\\code\\PM2.5\\BeijingPM20100101_20151231.csv")
      */

    /**
      * 2、直接读取字符串、然后继续map算子继续转换
      *spark.read.csv().map(row=>row)
      */
    //第三种
    val schema = StructType(
      Seq(
        StructField("No", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("pm", DoubleType)
      )
    )
    val source = spark.read
      .option("header", true)
      .schema(schema)
      .csv("E:\\PHY\\Pandas\\使用Pandas进行数据预处理\\resource\\BeijingPM_Nan.csv")

    //缺失值的删除
    //只要有一个Nan就删除,默认是any
    source.na.drop().show()
    //只有全是Nan才删除
    source.na.drop("all").show()
    //这个any'只作用于这四列
    source.na.drop("any", List("year", "month", "day", "hour")).show()
    //缺失值的删除
    //针对所有的
    source.na.fill(0).show()
    //只针对特定的列
    source.na.fill(0, List("month")).show()
  }

  @Test
  def strProcessor(): Unit = {
    //读取数据集
    val source = spark.read
      .option("header", true)
      .option("inferSchema", true)
      .csv("E:\\PHY\\Pandas\\使用Pandas进行数据预处理\\resource\\BeijingPM20100101_20151231.csv")
    //丢弃
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //不等于
    source.where('PM_Dongsi =!= "NA").show()
    //替换
    source.select(
      'No as "id", 'year, 'month, 'day, 'hour, 'season,
      when('PM_Dongsi === "NA", Double.NaN)
        .otherwise('PM_Dongsi cast DoubleType)
        .as("pm")
    ).show()
    //注意的是：原类型和转换的类型必须一致
    //将PM_Dongsi中的Na替换成NaN
    source.na.replace("PM_Dongsi",Map("NA" -> "NaN")).show()
  }
}
